package com.bs.service;

import com.bs.domain.Svc10MinutesExecuteInfo;
import com.bs.domain.SvcTop5ExecuteInfo;
import com.bs.service.callback.ModuleNameCallBack;
import com.bs.utils.DataRowCollection;
import com.bs.utils.DataTable;
import org.elasticsearch.action.search.SearchRequestBuilder;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Client;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.histogram.DateHistogramBuilder;
import org.elasticsearch.search.aggregations.bucket.histogram.InternalHistogram;
import org.elasticsearch.search.aggregations.bucket.terms.*;
import org.elasticsearch.search.aggregations.metrics.sum.InternalSum;
import org.elasticsearch.search.aggregations.metrics.sum.SumBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.ElasticsearchTemplate;
import org.springframework.stereotype.Service;

import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.stream.Collectors;

@Service
public class Svc10MinutesExecuteInfoService {

    @Autowired
    ElasticsearchTemplate elasticsearchTemplate;
    @Autowired
    private ModuleNameService moduleNameService;
    private String partition;
    private String startTime;
    private String endTime;
    private Map<String, String> svcSerMap = new HashMap<>();
    private Map<String, List<SvcTop5ExecuteInfo>> timeSrvMap = new HashMap<>();
    private StringBuilder svcNames = new StringBuilder();
    Map<SvcTop5ExecuteInfo, String> svcExecuteInfoMap = new HashMap<>();

    private SearchResponse getResponse() {
        Client transportClient = elasticsearchTemplate.getClient();
        BoolQueryBuilder builder = QueryBuilders.boolQuery();
        builder.must(QueryBuilders.matchPhrasePrefixQuery("logmsg", "4. service end."));
        builder.filter(QueryBuilders.boolQuery().must(QueryBuilders.termQuery("partition", partition))
            .must(QueryBuilders.rangeQuery("@timestamp").from(startTime).to(endTime).timeZone("+08:00")));
        SearchRequestBuilder searchRequestBuilder = transportClient.prepareSearch("logstash-*-svcsql-*").setTypes("log");
        DateHistogramBuilder timeAgg = AggregationBuilders.dateHistogram("srv_exec_time").field("@timestamp").interval(600000);
        TermsBuilder srvAgg = AggregationBuilders.terms("servers").field("pid").size(10).order(Terms.Order.aggregation("sum_srv_exec_time", false));
        SumBuilder sumSrvAgg = AggregationBuilders.sum("sum_srv_exec_time").field("exec_time");
        SumBuilder sumSvcAgg = AggregationBuilders.sum("sum_svc_exec_time").field("exec_time");
        TermsBuilder svcAgg = AggregationBuilders.terms("services").field("svc").order(Terms.Order.aggregation("sum_svc_exec_time", false));
        TermsBuilder hostAgg = AggregationBuilders.terms("hosts").field("host.raw");
        return searchRequestBuilder
            .setSize(0)
            .setQuery(builder)
            .addAggregation(timeAgg.subAggregation(srvAgg.subAggregation(sumSrvAgg).subAggregation(svcAgg.subAggregation(sumSvcAgg)).subAggregation(hostAgg)))
            .execute().actionGet();

    }

    public List<Svc10MinutesExecuteInfo> getExecuteInfo(String partition, String startTime, String endTime, String svcName) {
        List<Svc10MinutesExecuteInfo> svc10MinutesExecuteInfos = new ArrayList<>();
        this.partition = partition;
        this.startTime = startTime;
        this.endTime = endTime;
        SearchResponse response = getResponse();
        InternalHistogram timeAgg = response.getAggregations().get("srv_exec_time");
        List<InternalHistogram.Bucket> bucketList = timeAgg.getBuckets();
        bucketList.stream().forEach(bucket -> {
            String key = transDate(bucket.getKeyAsString());
           List<SvcTop5ExecuteInfo> serverExecuteInfoList = new ArrayList<>();
            svc10MinutesExecuteInfos.add(new Svc10MinutesExecuteInfo(key, bucket.getDocCount(), serverExecuteInfoList));
            InternalTerms srvAgg = bucket.getAggregations().get("servers");
            List<Terms.Bucket> srvBuckets = srvAgg.getBuckets();
            srvBuckets.stream().forEach(srvBucket -> {
                double sum = ((InternalSum) srvBucket.getAggregations().get("sum_srv_exec_time")).getValue();
                InternalTerms svcAgg = srvBucket.getAggregations().get("services");
                InternalTerms hostAgg = srvBucket.getAggregations().get("hosts");
                List<Terms.Bucket> hostBuck = (List<Terms.Bucket>) hostAgg.getBuckets();
                String hostName = hostBuck.size()>0?hostBuck.get(0).getKeyAsString():" ";
                String svc1 = getSvc(svcAgg, 0);
                String svc2 = getSvc(svcAgg, 1);
                String svc3 = getSvc(svcAgg, 2);
                String svc4 = getSvc(svcAgg, 3);
                String svc5 = getSvc(svcAgg, 4);
                String svc_name = svc1.split("/")[0];
                svcNames.append("'").append(svc_name).append("'").append(",");
                SvcTop5ExecuteInfo svcExecuteInfo = new SvcTop5ExecuteInfo(null, sum, svc1, svc2, svc3, svc4, svc5, hostName);
                svcExecuteInfoMap.put(svcExecuteInfo, svc_name);
                serverExecuteInfoList.add(svcExecuteInfo);
            });
            timeSrvMap.put(key,serverExecuteInfoList);
        });
        setServerName(svcExecuteInfoMap);

        if ("".equals(svcName))
            return svc10MinutesExecuteInfos;
        else
            return svc10MinutesExecuteInfos.stream().filter(svc10MinutesExecuteInfo -> svc10MinutesExecuteInfo.getServerList().stream()
                .filter(svcTop5ExecuteInfo -> svcName.equals(svcTop5ExecuteInfo.getSvc1().split("/")[0]) || svcName
                    .equals(svcTop5ExecuteInfo.getSvc2().split("/")[0]) || svcName.equals(svcTop5ExecuteInfo.getSvc3().split("/")[0]) || svcName
                    .equals(svcTop5ExecuteInfo.getSvc4().split("/")[0]) || svcName.equals(svcTop5ExecuteInfo.getSvc5().split("/")[0]))
                .count() != 0).collect(Collectors.toList());
    }


    private String transDate(String key) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss.SSSZ");
        Date parsedDate = null;
        try {
            parsedDate = dateFormat.parse(key.replaceAll("Z$", "+0000"));
        } catch (ParseException e) {
            e.printStackTrace();
        }
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        sdf.setTimeZone(TimeZone.getTimeZone("Asia/Shanghai"));
        return sdf.format(parsedDate);
    }

    public List<SvcTop5ExecuteInfo> getSvcTop5ExecuteInfo(String key) {
        return timeSrvMap.get(key);
    }


    private String getSvc(InternalTerms svcAgg, int index) {
        try {
            Terms.Bucket svcBucket = (Terms.Bucket) svcAgg.getBuckets().get(index);
            return svcBucket.getKeyAsString() + "/" + (int) (((InternalSum) svcBucket.getAggregations().get("sum_svc_exec_time")).getValue()) + "ms";
        } catch (IndexOutOfBoundsException e) {
            return " ";
        }

    }
    private void setServerName(Map<SvcTop5ExecuteInfo, String> serverExecuteInfoMap) {
        getSvcSvrMap();
        for (SvcTop5ExecuteInfo svcExecuteInfo : serverExecuteInfoMap.keySet()) {
            String svc_name = serverExecuteInfoMap.get(svcExecuteInfo);
            String ser_id = svcSerMap.get(svc_name);
            svcExecuteInfo.setServer((ser_id != null) ? ser_id : "EMPTY");
        }

    }



    private void getSvcSvrMap() {
        String svc_names = svcNames.toString();
        if (svc_names.length()>0&&svc_names.substring(svc_names.length()-1,svc_names.length()).trim().equals(",")){
            svc_names = svc_names.substring(0,svc_names.length()-1);
        }
        String sql = "SELECT S.SUB_SYSTEM_ENAME || '-' || S.SRV_ID AS SRV_ID, C.SVC_NAME FROM TEA00 S JOIN TEA01 C ON S.SRV_ID = C.SRV_ID\n" +
            "WHERE C.SVC_NAME IN (%s) ";
        moduleNameService.getModule(partition, sql, svc_names, new ModuleNameCallBack() {

            @Override
            public void onSuccess(DataTable table) {
                svcNames = new StringBuilder("");
                DataRowCollection rows = table.getRows();
                rows.forEach(dataRow -> svcSerMap.put(dataRow.getValue("SVC_NAME").toString(), dataRow.getValue("SRV_ID").toString()));
            }

            @Override
            public void onFailure(String msg) {
                System.out.println(msg);
            }
        });
    }
}
